// @Author : zaniu(zzaniu@126.com)
// @Time   : 2021/8/31 16:33
// @Desc   :
//
//                   _ooOoo_
//                  o8888888o
//                  88" . "88
//                  (| -_- |)
//                  O\  =  /O
//               ____/`---'\____
//             .'  \\|     |//  `.
//            /  \\|||  :  |||//  \
//           /  _||||| -:- |||||-  \
//           |   | \\\  -  /// |   |
//           | \_|  ''\---/''  |   |
//           \  .-\__  `-`  ___/-. /
//         ___`. .'  /--.--\  `. . __
//      ."" '<  `.___\_<|>_/___.'  >'"".
//     | | :  `- \`.;`\ _ /`;.`/ - ` : | |
//     \  \ `-.   \_ __\ /__ _/   .-` /  /
// =====`-.____`-.___\_____/___.-`____.-'=====
//                   `=---='
// ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
//           佛祖保佑       永无BUG
//
package client

import (
	"context"
	"distributedTransaction/client/conf"
	"distributedTransaction/proto2go"
	"distributedTransaction/utils"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"os"
	"time"
)

type DemoListener struct{}

func NewDemoListener() *DemoListener {
	return &DemoListener{}
}

func (dl *DemoListener) ExecuteLocalTransaction(msg *primitive.Message) primitive.LocalTransactionState {
	msgInfo := utils.MsgInfo{}
	err := json.Unmarshal(msg.Body, &msgInfo)
	if err != nil {
		fmt.Println("消息解析失败, err = ", err.Error())
		return primitive.CommitMessageState
	}
	rpcOrderServerInfo, err := conf.GetOneRpcServerInfo(conf.GlobalOrderServices)
	if err != nil {
		panic(err)
	}
	orderConn, err := grpc.Dial(fmt.Sprintf("%s:%d", rpcOrderServerInfo.Ip, rpcOrderServerInfo.Port), grpc.WithInsecure())
	if err != nil {
		panic("grpc.Dial err:" + err.Error())
	}
	defer orderConn.Close()
	rpcInvenServerInfo, err := conf.GetOneRpcServerInfo(conf.GlobalInventoryServices)
	if err != nil {
		panic(err)
	}
	invenConn, err := grpc.Dial(fmt.Sprintf("%s:%d", rpcInvenServerInfo.Ip, rpcInvenServerInfo.Port), grpc.WithInsecure())
	if err != nil {
		panic("grpc.Dial err:" + err.Error())
	}
	defer invenConn.Close()
	err = detainInventory(invenConn, msgInfo.GlobalNo, msgInfo.UserName, msgInfo.GoodsName, msgInfo.GoodsNumber)
	if err != nil {
		st, ok := status.FromError(err)
		// 如果不是OK的话，要回滚消息，说明库存扣减失败，直接不处理
		if !(ok && st.Code() == codes.OK) {
			return primitive.CommitMessageState
		}
	}
	// 这里去创建订单，如果订单创建成功的话就CommitMessageState， 创建失败就RollbackMessageState
	err = CreateOrder(orderConn, msgInfo.GlobalNo, msgInfo.UserName, msgInfo.GoodsName, msgInfo.GoodsNumber)
	if err != nil {
		st, ok := status.FromError(err)
		// 如果是没有找到的话，直接提交消息去回滚库存就OK了
		if ok && st.Code() == codes.Unknown {
			// 成功的情况直接 rollback 消息就行了
			return primitive.UnknowState
		} else {
			// 其他的错误我也懒的去一个一个搞了，直接让回查吧，回查如果没有查到的话会去回滚库存的
			return primitive.CommitMessageState
		}
	}
	return primitive.RollbackMessageState
}

func (dl *DemoListener) CheckLocalTransaction(msg *primitive.MessageExt) primitive.LocalTransactionState {
	// 需要调用库存和订单接口，如果发现订单有，直接RollbackMessageState，不处理就行。订单没有的话就需要CommitMessageState
	// 如果一直是UnknowState的话，默认会回查16次，然后回滚该消息
	fmt.Printf("回查消息， %v msg transactionID : %v\n", time.Now(), msg.TransactionId)
	m := utils.MsgInfo{}
	json.Unmarshal(msg.Body, &m)
	rpcOrderServerInfo, err := conf.GetOneRpcServerInfo(conf.GlobalOrderServices)
	if err != nil {
		panic(err)
	}
	orderConn, err := grpc.Dial(fmt.Sprintf("%s:%d", rpcOrderServerInfo.Ip, rpcOrderServerInfo.Port), grpc.WithInsecure())
	if err != nil {
		panic("grpc.Dial err:" + err.Error())
	}
	defer orderConn.Close()
	orderClient := proto2go.NewOrderClient(orderConn)
	queryOrderRequest := &proto2go.QueryOrderExistsRequest{GlobalNo: m.GlobalNo}
	queryOrderExistsRequest, err := orderClient.QueryOrderExists(context.Background(), queryOrderRequest)
	if err != nil {
		st, ok := status.FromError(err)
		// 如果是没有找到的话，直接提交消息去回滚库存就OK了
		if ok && st.Code() == codes.NotFound {
			return primitive.CommitMessageState
		} else {
			return primitive.UnknowState
		}
	}
	// 订单存在，回滚消息. 如果不存在, 提交消息去回滚库存
	if queryOrderExistsRequest.Exists {
		return primitive.RollbackMessageState
	} else {
		return primitive.CommitMessageState
	}
}

// 这个应该像rabbitmq一样是全局的啊，不应该只是局部的
func StartTransactionProducer() *rocketmq.TransactionProducer {
	p, _ := rocketmq.NewTransactionProducer(
		NewDemoListener(),
		// NameServer 是个集群嘛，所以IP可以是多个
		producer.WithNsResolver(primitive.NewPassthroughResolver([]string{fmt.Sprintf("%s:%d", utils.GlobalServerConfig.RocketMQHost, utils.GlobalServerConfig.RocketMQPort)})),
		producer.WithRetry(1),
	)
	err := p.Start()
	if err != nil {
		fmt.Printf("start producer error: %s\n", err.Error())
		os.Exit(1)
	}
	return &p
}

// 发送事务消息
func SendTransactionMessage(p *rocketmq.TransactionProducer, globalNo, userName, goodsName string, goodsNumber uint64) error {
	msg, err := json.Marshal(utils.MsgInfo{GlobalNo: globalNo, UserName: userName, GoodsName: goodsName, GoodsNumber: goodsNumber})
	if err != nil {
		return err
	}
	// 这个里面是同步发送的 ExecuteLocalTransaction. 为啥是同步啊...我醉了
	// TODO 这里有点问题啊，这里 ExecuteLocalTransaction 里面的错误不能传出来。比如RPC调用失败或者超时是传不出来的。难道不能在这个里面搞？
	resp, err := (*p).SendMessageInTransaction(context.Background(), primitive.NewMessage(utils.GlobalServerConfig.Topic, msg))
	if err != nil {
		return err
	}
	if resp.State != primitive.RollbackMessageState {
		return errors.New("下单失败，可能是库存不足")
	}
	return nil
}

var Rocket *rocketmq.TransactionProducer

//
// func init() {
// 	Rocket = StartTransactionProducer()
// }
